package xin.marcher.module.message.mq.consumer;

import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.common.protocol.heartbeat.MessageModel;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import xin.marcher.module.common.constants.RocketMqConstant;
import xin.marcher.module.message.mq.config.RocketMQProperties;
import xin.marcher.module.message.mq.consumer.listener.DataChangeListener;
import xin.marcher.module.message.mq.consumer.listener.DataExternalChangeListener;
import xin.marcher.module.message.mq.consumer.listener.DataMessageCacheListener;
import xin.marcher.module.message.mq.consumer.listener.DataTraceListener;

import javax.annotation.Resource;

/**
 *
 */
@Configuration
public class ConsumerBeanConfig {

    /**
     * 配置内容对象
     */
    @Resource
    private RocketMQProperties rocketMQProperties;

    /**
     * 数据变更消费者，负责数据变更监控后发送MQ消息通知订阅方
     *
     * @param dataChangeListener
     * @return
     * @throws MQClientException
     */
    @Bean("dataChangeConsumer")
    public DefaultMQPushConsumer createItemStageConsumer(DataChangeListener dataChangeListener) throws MQClientException {
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer(RocketMqConstant.DATA_CHANGE_CONSUMER_GROUP);
        consumer.setInstanceName("dataChangeListener");
        consumer.setMessageModel(MessageModel.BROADCASTING);
        consumer.setNamesrvAddr(rocketMQProperties.getNameServer());
        consumer.subscribe(RocketMqConstant.DATA_CHANGE_TOPIC, "*");
        consumer.registerMessageListener(dataChangeListener);
        consumer.start();
        return consumer;
    }

    /**
     * 数据变更消费者，负责记录数据变更
     *
     * @param dataTraceListener
     * @return
     * @throws MQClientException
     */
    @Bean("dataTraceConsumer")
    public DefaultMQPushConsumer dataTraceTopic(DataTraceListener dataTraceListener) throws MQClientException {
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer(RocketMqConstant.DATA_CHANGE_CONSUMER_GROUP);
        consumer.setInstanceName("dataTraceListener");
        consumer.setMessageModel(MessageModel.BROADCASTING);
        consumer.setNamesrvAddr(rocketMQProperties.getNameServer());
        consumer.subscribe(RocketMqConstant.DATA_CHANGE_TOPIC, "*");
        consumer.registerMessageListener(dataTraceListener);
        consumer.start();
        return consumer;
    }

    /**
     * 数据变更消费者，负责数据变更监控后发送MQ消息通知订阅方
     *
     * @param dataExternalChangeListener
     * @return
     * @throws MQClientException
     */
    @Bean("dataExternalChangeConsumer")
    public DefaultMQPushConsumer dataExternalChangeTopic(DataExternalChangeListener dataExternalChangeListener) throws MQClientException {
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer(RocketMqConstant.DATA_EXTERNAL_CHANGE_CONSUMER_GROUP);
        consumer.setNamesrvAddr(rocketMQProperties.getNameServer());
        consumer.subscribe(RocketMqConstant.DATA_EXTERNAL_CHANGE_TOPIC, "*");
        consumer.registerMessageListener(dataExternalChangeListener);
        consumer.start();
        return consumer;
    }


    /**
     * 负责监听缓存消息的变化，对应处理具体的缓存操作
     *
     * @param dataMessageCacheListener
     * @return
     * @throws MQClientException
     */
    @Bean("dataMessageCacheConsumer")
    public DefaultMQPushConsumer dataMessageCacheTopic(DataMessageCacheListener dataMessageCacheListener) throws MQClientException {
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer(RocketMqConstant.DATA_MESSAGE_CACHE_CONSUMER_GROUP);
        consumer.setInstanceName("dataMessageCacheListener");
        consumer.setNamesrvAddr(rocketMQProperties.getNameServer());
        consumer.subscribe(RocketMqConstant.DATA_MESSAGE_CACHE_TOPIC, "*");
        consumer.registerMessageListener(dataMessageCacheListener);
        consumer.start();
        return consumer;
    }
}
